package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.DimFunctionWithCache;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/2/20 09:57
 */
public class Dws_09_DwsTradeSkuOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_09_DwsTradeSkuOrderWindow().init(
            4009,
            2,
            "Dws_09_DwsTradeSkuOrderWindow",
            Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        // 1. 先封装数据到 pojo
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream = parseToPojo(stream);
        // 2. 按照 order_detail_id 实现去重
        beanStream = distinctByOrderDetailId(beanStream);
        // 3. 开窗聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> resultStreamWithoutDimsStream = windowAndAgg(beanStream);
        // 5. 补充维度信息
        SingleOutputStreamOperator<TradeSkuOrderBean> resultStream = joinDims(resultStreamWithoutDimsStream);
        resultStream.print();
    
        // 6. 写出到 clickhouse 中
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> joinDims(SingleOutputStreamOperator<TradeSkuOrderBean> stream) {
        /*
        
        sku_id   amount1 amount2   sku_name  spu_id  spu_name  ..
        1          100    200
        
        select * from dim_sku_info where id='1'
        select * from dim_spu_info where id=?
        
     
         */
        SingleOutputStreamOperator<TradeSkuOrderBean> skuInfoStream = stream
            .map(new DimFunctionWithCache<TradeSkuOrderBean>() {
                @Override
                public String getTable() {
                    return "dim_sku_info";
                }
            
                @Override
                public Object getId(TradeSkuOrderBean bean) {
                    return bean.getSkuId();
                }
            
                @Override
                public void addDim(TradeSkuOrderBean value, JSONObject dim) {
                    value.setSkuName(dim.getString("SKU_NAME"));
                    value.setSpuId(dim.getString("SPU_ID"));
                    value.setCategory3Id(dim.getString("CATEGORY3_ID"));
                    value.setTrademarkId(dim.getString("TM_ID"));
                }
            });
    
        SingleOutputStreamOperator<TradeSkuOrderBean> spuInfoStream = skuInfoStream
            .map(new DimFunctionWithCache<TradeSkuOrderBean>() {
                @Override
                public String getTable() {
                    return "dim_spu_info";
                }
            
                @Override
                public Object getId(TradeSkuOrderBean bean) {
                    return bean.getSpuId();
                }
            
                @Override
                public void addDim(TradeSkuOrderBean value, JSONObject dim) {
                    value.setSpuName(dim.getString("SPU_NAME"));
                    
                }
            });
    
        SingleOutputStreamOperator<TradeSkuOrderBean> tmStream = spuInfoStream
            .map(new DimFunctionWithCache<TradeSkuOrderBean>() {
                @Override
                public String getTable() {
                    return "dim_base_trademark";
                }
            
                @Override
                public Object getId(TradeSkuOrderBean bean) {
                    return bean.getTrademarkId();
                }
            
                @Override
                public void addDim(TradeSkuOrderBean value, JSONObject dim) {
                    value.setTrademarkName(dim.getString("TM_NAME"));
                
                }
            });
    
    
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = tmStream
            .map(new DimFunctionWithCache<TradeSkuOrderBean>() {
                @Override
                public String getTable() {
                    return "dim_base_category3";
                }
            
                @Override
                public Object getId(TradeSkuOrderBean bean) {
                    return bean.getCategory3Id();
                }
            
                @Override
                public void addDim(TradeSkuOrderBean value, JSONObject dim) {
                    value.setCategory3Name(dim.getString("NAME"));
                    value.setCategory2Id(dim.getString("CATEGORY2_ID"));
                
                }
            });
    
    
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = c3Stream
            .map(new DimFunctionWithCache<TradeSkuOrderBean>() {
                @Override
                public String getTable() {
                    return "dim_base_category2";
                }
            
                @Override
                public Object getId(TradeSkuOrderBean bean) {
                    return bean.getCategory2Id();
                }
            
                @Override
                public void addDim(TradeSkuOrderBean value, JSONObject dim) {
                    value.setCategory2Name(dim.getString("NAME"));
                    value.setCategory1Id(dim.getString("CATEGORY1_ID"));
                
                }
            });
    
        
        return c2Stream
            .map(new DimFunctionWithCache<TradeSkuOrderBean>() {
                @Override
                public String getTable() {
                    return "dim_base_category1";
                }
        
                @Override
                public Object getId(TradeSkuOrderBean bean) {
                    return bean.getCategory1Id();
                }
        
                @Override
                public void addDim(TradeSkuOrderBean value, JSONObject dim) {
                    value.setCategory1Name(dim.getString("NAME"));
                }
            });
    
    
    
    
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> windowAndAgg(
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream) {
        return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(TradeSkuOrderBean::getSkuId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1,
                                                    TradeSkuOrderBean value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                        value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                        
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String skuId,
                                        Context ctx,
                                        Iterable<TradeSkuOrderBean> elements,
                                        Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(ctx.window().getEnd()));
                        
                        bean.setTs(System.currentTimeMillis());
                        
                        out.collect(bean);
                    }
                }
            );
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> distinctByOrderDetailId(
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream) {
        return beanStream
            .keyBy(TradeSkuOrderBean::getOrderDetailId)
            .map(new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                
                private ValueState<TradeSkuOrderBean> lastBeanState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 给状态设置 ttl
                    StateTtlConfig config = new StateTtlConfig
                        .Builder(org.apache.flink.api.common.time.Time.seconds(10)) // ttl 时间
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite) // 当创建写和读的时候,更新 ttl
                        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) // 当过期之后, 如果去获取状态返回 null
                        .build();
                    
                    ValueStateDescriptor<TradeSkuOrderBean> desc = new ValueStateDescriptor<>("lastBean", TradeSkuOrderBean.class);
                    desc.enableTimeToLive(config);
                    
                    lastBeanState = getRuntimeContext().getState(desc);
                }
                
                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean bean) throws Exception {
                    // 当第一条数据来的时候: 1. 把数据存储到状态中 2. 输出
                    
                    // 不是第一条: 1. 对四个指标用新值-状态中的值, 输出  2. 把第二条数据存入到状态
                    
                    TradeSkuOrderBean lastBean = lastBeanState.value();
                    if (lastBean != null) {
                        // 计算后的值,存入到 lastBean, 把 lastBean 输出
                        lastBean.setOrderAmount(bean.getOrderAmount().subtract(lastBean.getOrderAmount()));  // 新值 - 旧值
                        lastBean.setOriginalAmount(bean.getOriginalAmount().subtract(lastBean.getOriginalAmount()));  // 新值 - 旧值
                        lastBean.setActivityAmount(bean.getActivityAmount().subtract(lastBean.getActivityAmount()));  // 新值 - 旧值
                        lastBean.setCouponAmount(bean.getCouponAmount().subtract(lastBean.getCouponAmount()));  // 新值 - 旧值
                    }
                    lastBeanState.update(bean);  // 新值存入到状态中
                    return lastBean == null ? bean : lastBean; // 计算后的值
                }
            });
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> parseToPojo(DataStreamSource<String> stream) {
        return stream.map(new MapFunction<String, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(String value) throws Exception {
                JSONObject obj = JSON.parseObject(value);
                
                return TradeSkuOrderBean.builder()
                    .orderDetailId(obj.getString("id"))
                    .skuId(obj.getString("sku_id"))
                    .orderAmount(obj.getBigDecimal("split_total_amount"))
                    .originalAmount(obj.getBigDecimal("split_original_amount"))
                    .activityAmount(obj.getBigDecimal("split_activity_amount") == null ? new BigDecimal(0) : obj.getBigDecimal("split_activity_amount"))
                    .couponAmount(obj.getBigDecimal("split_coupon_amount") == null ? new BigDecimal(0) : obj.getBigDecimal("split_coupon_amount"))
                    .ts(obj.getLong("ts") * 1000)  // ts变成毫秒
                    .build();
            }
        });
    }
}
/*
交易域SKU粒度 下单各窗口汇总表
原始金额、活动减免金额、优惠券减免金额和订单金额，

数据源: dwd 层下单事务事实表


order_detaitl_id  sku_id   原始金额    activity  coupon    time
	1				2       100        null     null       0
	null  (忽略)
	1 				2	    100        xx       null       1
	1 				2	    100        xx        yy        2
	
select sum(原始金额) from a group by sku_id

1. 按照详情 id 进行去重, 保留最完整的数据
    keyBy -> order_detail_id 分组
    
    a: 窗口 session: gap 5s
    
        process 拿到所有的数据(<=3)
            找到时间最大的
       
       什么时候可以输出结果?
           最后一条数据到了 5s 之后, 结果出来
           
       缺点: 时效低
       
    b: 第一条数据到了之后, 5s 后,最后一条也一定到了. 这个找到时间最大的
    
        定时器:
            第一条数据来了之后, 注册一个 5s 的定时器,定时器触发的时候, 表示数据来齐了. 计算结果
            
          什么时候可以输出结果?
           第一条数据到了 5s 之后, 结果出来
       实效性高于窗口
       
    c: 如果需要的数据都在左表, 则可以直接输出第一条数据, 后面的不要了.
    
        实效性很高
        
    
    d: 补偿法
    
    order_detaitl_id  sku_id     原始金额    activity   coupon    time
	1				   2           100        0         0       0            => 输出
	null  (忽略)
	
	1				   2           -100       0          0           0       => 上条数据取反
	1 				   2	       100        200       0        1           => 输出
	
	1 				   2	       -100       -200     0        1            => 上条数据取反
	1 				   2	       100        200      300        2          => 输出
	
	时效性高: 不需要等待
	    正常是 3 条数据, 实际向流中写出了 5 条: 写放大
	 
	 
	 order_detaitl_id  sku_id     原始金额    activity   coupon    time
	1				   2           100        0         0       0            => 输出
	null  (忽略)
	
	1 				   2	       100-100   200-0       0-0        1        => 输出
	
	1 				   2	       100      200            0        1
	1 				   2	       100-100  200 200     300-0        2       => 输出
	  没有写放大, 时效性高
	
	
	
    
    
    
    
    
2. 封装到 pojo 中

3. 按照 sku_id 分组 开窗聚和

4. 和 sku 相关的维度都应该拿到: spu tm c3 c2 c1
    事实表与维度表的 join
    
        需要用到手动 join
            sku_id=1     select * from dim_sku_info where id=1

5. 写出到 clickhouse 中
   

 */